package com.guchenbo.spark.sql

import org.apache.spark.SparkConf

/**
 * @author guchenbo
 * @date 2021/6/25
 */
object OracleReader {

  def main(args: Array[String]): Unit = {
    //    yarn()
    val conf = new SparkConf().setAppName("test-sql").setMaster(master).setJars(jars)
    val spark = org.apache.spark.sql.SparkSession.builder().config(conf).getOrCreate()

    read(spark)
  }

  var master: String = "local"
  var jars: Array[String] = Array("")

  def yarn(): Unit = {
    master = "yarn"
    jars = Array("hdfs:///user/turing/algorithm/prod/modelpaas/algoso/model-pass-data-exchange-1-0-3.jar")
  }

  def read(spark: org.apache.spark.sql.SparkSession): Unit = {
    println("read table")
    val src = "GMT_CREATE"
    val dbtype = "TIMESTAMP"
    //    val dbtype = "DATE"

    var sql = "select * from SPARTAN.BIFROST_USER"
    sql = "with view_alias as (select * from SPARTAN.BIFROST_USER) select * from (select * from ( select * from view_alias ) Table_0) dims"

    val df = spark.read
      .format("jdbc")
      .option("url", "jdbc:oracle:thin:@10.57.34.44:1521:ORCL")
      .option("user", "kratos")
      .option("password", "12345678")
      .option("driver", "oracle.jdbc.OracleDriver")
      .option("dbtable", s"($sql) sub")
      .load()

    df.show()
//    val fns = df.schema.fieldNames
//    var map: Map[String, Integer] = Map()
//    for (i <- fns.indices) map += (fns(i) -> i)
//    var col = ""
//    var idx = map(col)
//    val colConf = Map("transform" -> src, "transformType" -> dbtype)
//    val writeMaps = Map(src -> colConf)
//    val rdd = df.rdd.map(row => mapMethod(row, fns, writeMaps))
//    val array = rdd.collect()
//
//    println("find data size " + array.length)
//    println(array.mkString("Array(", ", ", ")"))
  }

  def mapMethod(row: org.apache.spark.sql.Row, fns: Array[String], writeMaps: Map[String, Map[String, String]]): java.util.Date = {
    var date: java.util.Date = null

    for (i <- fns.indices) {
      val name = fns(i)
      if (writeMaps.contains(name)) {
        val conf = writeMaps(name)
        val dbt = conf("transformType")
        if (dbt.equalsIgnoreCase("date")) {
          try {
            date = row.getDate(i)
          } catch {
            case exception: Exception =>
              println("getDate error then getTimestamp", exception)
              date = row.getTimestamp(i)
          }
        } else {
          try {
            date = row.getTimestamp(i)
          } catch {
            case exception: Exception =>
              println("getTimestamp error then getDate", exception)
              date = row.getDate(i)
          }
        }
        if (null == date) {
          println("date is null")
        }
      }
    }
    date
  }
}
